package spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object updateStateByKey {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("NetworkWordCount")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // checkPoint
    ssc.checkpoint("./ck")

    // Create a DStream that will connect to hostname:port, like hadoop102:9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("linux1", 9999)

    // Split each line into words
    val words: DStream[String] = lines.flatMap(_.split(" "))


    //import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
    // Count each word in each batch
    val pairs: DStream[(String, Int)] = words.map(word => (word, 1))


    // 使用updateStateByKey来更新状态，统计从运行开始以来单词总的次数
    val stateDStream: DStream[(String, Int)] = pairs.updateStateByKey[Int](updateFunc)


    stateDStream.print()

    // Start the computation and  Wait for the computation to terminate
    ssc.start()
    ssc.awaitTermination()
  }

  // 定义更新状态方法，参数values为当前批次单词频度，state为以往批次单词频度
  private val updateFunc: (Seq[Int], Option[Int]) => Some[Int] = (values: Seq[Int], state: Option[Int]) => {

    val currentCount: Int = values.sum
    val previousCount: Int = state.getOrElse(0)

    Some(currentCount + previousCount)
  }
}
